package cn.itcast.streaming.task;

import cn.itcast.streaming.entity.ElectricFenceModel;
import cn.itcast.streaming.entity.ElectricFenceResultTmp;
import cn.itcast.streaming.entity.ItcastDataPartObj;
import cn.itcast.streaming.flatmap.ElectricFenceModelFunction;
import cn.itcast.streaming.flatmap.ElectricFenceRulesFuntion;
import cn.itcast.streaming.sink.ElectricFenceMysqlSink;
import cn.itcast.streaming.source.MySQLElectricFenceSouce;
import cn.itcast.streaming.source.MysqlElectricFenceResultSource;
import cn.itcast.streaming.utils.JsonParsePartUtil;
import cn.itcast.streaming.window.function.ElectricFenceWindowFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.datanucleus.util.StringUtils;

import java.util.HashMap;

/**
 * Author itcast
 * Date 2021/9/26 10:57
 * Desc 电子围栏，用于判断实时车辆上报数据和电子围栏的规则静态数据进行关联，分析车辆的在指定的
 * 电子栅栏的信息和进入栅栏和出栅栏的时间入库。
 * 分析步骤：
 * 1. 电子围栏分析任务设置、原始数据json解析、过滤异常数据
 * 2. 读取已存在电子围栏中的车辆与电子围栏信息(广播流临时结果数据)
 * 3. 原始车辆数据与电子围栏广播流进行合并，生成电子围栏规则模型流数据（DStream<ElectricFenceModel>）
 * 4. 创建90秒翻滚窗口，计算电子围栏信息(ElectricFenceModel中的值根据车辆是否在围栏内进行设置)
 * 5. 读取电子围栏分析结果表数据并广播
 * 6. 翻滚窗口电子围栏对象模型流数据与电子围栏分析结果数据广播流进行connect
 * 7. 对电子围栏对象模型，添加uuid和inMySQL(车辆是否已存在mysql表中)
 * 8. 电子围栏分析结果数据落地mysql，也可以选择落地mongo
 */
public class ElectricFenceTask extends BaseTask {
    public static void main(String[] args) throws Exception {
        //1. 电子围栏分析任务设置、原始数据json解析、过滤异常数据
        StreamExecutionEnvironment env = getEnv(ElectricFenceTask.class.getSimpleName());
        DataStreamSource<String> kafkaStream = getKafkaStream(env, "__consumer_electricFence_", SimpleStringSchema.class);
        DataStream<ItcastDataPartObj> source = kafkaStream.map(JsonParsePartUtil::parseJsonToObject)
                .filter(obj -> StringUtils.isEmpty(obj.getErrorData()));
        //2. 读取已存在电子围栏中的车辆与电子围栏信息(广播流临时结果数据)
        DataStream<HashMap<String, ElectricFenceResultTmp>> electricFenceBroadcastStream = env
                .addSource(new MySQLElectricFenceSouce()).broadcast();
        //3. 原始车辆数据与电子围栏广播流进行合并，生成电子围栏规则模型流数据（DStream<ElectricFenceModel>）
        DataStream<ElectricFenceModel> electricFenceWidthStream = source.connect(electricFenceBroadcastStream)
                .flatMap(new ElectricFenceRulesFuntion());
        //4. 创建90秒翻滚窗口，计算电子围栏信息(ElectricFenceModel中的值根据车辆是否在围栏内进行设置)
        SingleOutputStreamOperator<ElectricFenceModel> electricFenceWindowStream = electricFenceWidthStream
                .assignTimestampsAndWatermarks(
                new BoundedOutOfOrdernessTimestampExtractor<ElectricFenceModel>(Time.seconds(3)) {
                    @Override
                    public long extractTimestamp(ElectricFenceModel element) {
                        return element.getTerminalTimestamp();
                    }
                }
        ).keyBy(obj -> obj.getVin())
                .window(TumblingEventTimeWindows.of(Time.seconds(90)))
                .apply(new ElectricFenceWindowFunction());
        //5. 读取电子围栏分析结果表数据并广播
        DataStream<HashMap<String, Integer>> electricFenceResultStream = env
                .addSource(new MysqlElectricFenceResultSource()).broadcast();
        //6. 翻滚窗口电子围栏对象模型流数据与电子围栏分析结果数据广播流进行connect
        electricFenceWindowStream.connect(electricFenceResultStream)
                .flatMap(new ElectricFenceModelFunction())
        //7. 对电子围栏对象模型，添加uuid和inMySQL(车辆是否已存在mysql表中)
        .addSink(new ElectricFenceMysqlSink());
        //8. 电子围栏分析结果数据落地mysql，也可以选择落地mongo
        //执行流环境
        env.execute();
    }
}
